package com.wk.springboot.bean;

import com.alibaba.fastjson.JSONObject;
import com.carrotsearch.hppc.ObjectLookupContainer;
import com.wk.springboot.bean.ExportEntity.field;
import com.wk.springboot.util.EsClientUtils;
import com.wk.springboot.writer.MyFileWriter;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.lucene.search.Query;
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.lucene.search.Queries;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryParseContext;
import org.elasticsearch.index.query.QueryShardContext;
import org.elasticsearch.index.query.SimpleQueryStringBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.SearchModule;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Classname EsExportRealBean
 * @Description 真正用到的代码
 * @Date 2020/5/30 15:57
 * @Author by weikai
 */
//@Configuration
public class EsExportRealBean implements InitializingBean {


  private final Logger LOG = LoggerFactory.getLogger(EsExportRealBean.class);


  private ExportEntity exportEntity;

  private String clusterName;

  private String ip;

  private int port = 9300;

  private String user;

  private String index;

  private String type;

  private String jsonPath;

  private Map<String, field> fields;

  private boolean trans;

  public static final String ONETOONE = "one2one";


  /**
   * 来自 ExportEntity 的字符串resultDefine, 或者为空时 来自 queryJSON 在 startExport 开始调用时初始化
   */
  private ResultDefine resultDefine;

  /**
   * 通过指定的QSL查询json,从ES库中查询指定索引和类型中的数据,导出具有一对一的扁平结构的数据
   */

  public void startExport(String writeFilePath, QueryBuilder query, String esDataType,
      boolean append,
      String exportType, boolean isZL) throws Exception {
    /* 可以解决 netty 的连接报错导致ES无法连接的问题(availableProcessors .... 之类),  */
    System.setProperty("es.set.netty.runtime.available.processors", "false");

    LOG.info("ElasticClient clusterName[{}] connecting...", clusterName);
    MyFileWriter writer = null;

    try {

      TransportClient client = EsClientUtils.createClient();;

      // 创建写入对象
      writer = new MyFileWriter(writeFilePath, append);
      // 分批查询数据并写入文件
      SearchResponse responseOfData;
      if (isZL) {
        responseOfData = getScollZL(query, client);
      } else {
        responseOfData = getScrollID(query, client);
      }
      // 查询第一个分页的ID
      SearchHits firstPageHit = responseOfData.getHits();
      //只导出为es的数据
      writeJsonToFile(firstPageHit, responseOfData, writer, client);

      // 将最后一批数据写入文件
      writer.close();
    } catch (Exception e) {
      throw new Exception("IOException ", e);
    } finally {
      if (writer != null) {
        writer.close();
      }
    }
  }


  /**
   * 查询第一个分页ID
   */
  private SearchResponse getScrollID(QueryBuilder query, TransportClient client) throws Exception {

    //version 5
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
    SearchModule searchModule = new SearchModule(Settings.EMPTY, false, Collections.emptyList());

    //try()在()中内容执行后会自动关闭
    try (
        XContentParser parser = XContentFactory.xContent(XContentType.JSON)
            .createParser(new NamedXContentRegistry(searchModule.getNamedXContents()),
                "{}")
    ) {
      sourceBuilder.parseXContent(new QueryParseContext(parser));
    } catch (IOException e) {
      LOG.error("获取分页信息失败");
      throw new Exception("获取分页信息失败", e);
    }

    SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client,
        SearchAction.INSTANCE)
        .setSource(sourceBuilder)
        .setScroll(new TimeValue(6000))
        .setSearchType(SearchType.QUERY_THEN_FETCH)
        .setSize(1000)
        .setIndices(index);

    if (StringUtils.isNotBlank(type)) {
      searchRequestBuilder.setTypes(type);
    }

    // -----获取分页的响应信息
    return searchRequestBuilder.execute().actionGet();
  }

  /**
   *
   * 查询增量数据
   */
  private SearchResponse getScollZL(QueryBuilder query, TransportClient client) throws Exception {
    SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client,
        SearchAction.INSTANCE)
        .setIndices(index)
        .setSearchType(SearchType.QUERY_THEN_FETCH)
        .setQuery(query)
        .setScroll(TimeValue.timeValueSeconds(6))
        .setSize(1000);


    if (StringUtils.isNotBlank(type)) {
      searchRequestBuilder.setTypes(type);
    }

    System.out.println("查询语句为"+searchRequestBuilder.toString());

    // -----获取分页的响应信息
    return searchRequestBuilder.execute().actionGet();
  }

  /**
   * 向文件中写入数据
   */
  private void writeJsonToFile(SearchHits firstPageHit, SearchResponse responseOfData,
      MyFileWriter writer, TransportClient client) throws Exception, InterruptedException {
    try {
      System.out.println("第一次查询的条数为："+firstPageHit.getHits().length);
      if (firstPageHit.getHits().length != 0) {
        extractLineData(firstPageHit, writer);
      }
      while (true) {
//        MonitorHelper.checkIn();
        responseOfData = client.prepareSearchScroll(responseOfData.getScrollId())
            .setScroll(TimeValue.timeValueSeconds(6)).execute().actionGet();
        SearchHits searchHit = responseOfData.getHits();
        // 当查询不到数据时跳出循环
        if (searchHit.getHits().length == 0) {
          break;
        }
        System.out.println("以后每次查询的条数为"+searchHit.getHits().length);
        extractLineData(searchHit, writer);
      }
    } catch (Exception e) {
      LOG.error("写入JSON文件错误");
      throw new Exception("写入JSON文件错误", e);
    }
  }

  /**
   * 往文件中hit
   */
  private void extractLineData(SearchHits firstPageHit, MyFileWriter writer) throws Exception {
    for (SearchHit hit : firstPageHit) {
      String jsonLineData = JSONObject.parseObject(hit.getSourceAsString()).toString() + "\r\n";
      writer.writeData(jsonLineData);
    }
  }

  /**
   * 创建query
   */
  private QueryBuilder boolQueryBuilder() {
    BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
    queryBuilder.filter(QueryBuilders.rangeQuery("lrsj").from("2015-07-01"));
    queryBuilder.filter(QueryBuilders.termQuery("name","王二丫2"));
    return queryBuilder;
  }


  public Map getMapping(String indexName, String indexType, Client client) throws Exception {

    ClusterState cs = client.admin().cluster().prepareState().setIndices(indexName).execute()
        .actionGet().getState();

    IndexMetaData imd = cs.getMetaData().index(indexName);

    ImmutableOpenMap<String, MappingMetaData> immutableOpenMap = imd.getMappings();

    ObjectLookupContainer<String> keys = immutableOpenMap.keys();

    MappingMetaData mdd = imd.mapping(indexType);

    return mdd.sourceAsMap();

  }

  @Override
  public void afterPropertiesSet() throws Exception {
    this.index = "i_rhk_np";
    this.type = "t_bysaj_yla";

    String filePath = "S:\\工作目录\\es\\yltest.dat";
//    String query = "{}";
    startExport(filePath,boolQueryBuilder(),"",true,"",false);
//    startExport(filePath, queryBuilder(), "", true, "", false);
  }
}
